package com.geniusk.mqtt.strategy.impl;

import com.geniusk.mqtt.TopicChannelSession;
import com.geniusk.mqtt.strategy.MessageStrategy;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Subscribe Ack消息
 */
@Component
public class SubscribeAckMessageStrategy  implements MessageStrategy {
    @Override
    public void sendResponseMessage(Channel channel, MqttMessage mqttMessage) {
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        //解析接收的消息
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
        List<Integer> grantedQoSLevels = new ArrayList<>();
        List<MqttTopicSubscription> mqttTopicSubscriptionList = mqttSubscribeMessage.payload().topicSubscriptions();
        for(MqttTopicSubscription mqttTopicSubscription:mqttTopicSubscriptionList){
            String topicName = mqttTopicSubscription.topicName();
            grantedQoSLevels.add(mqttTopicSubscription.qualityOfService().value());
            TopicChannelSession.setChannel(topicName, channel);
        }
        //构建返回的消息，有效负载
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2+topics.size());
        MqttSubAckMessage subAckMsg = new MqttSubAckMessage(mqttFixedHeaderBack,variableHeaderBack, payloadBack);
        channel.writeAndFlush(subAckMsg);

    }
}
